package com.li.flink.datastream

import com.li.flink.utils.MysqlUtils
import org.apache.flink.api.scala._
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

import java.sql.{Connection, PreparedStatement}

/**
 * 自定义mysql sink
 */
object Sink2Mysql {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    var connection: Connection = null
    var prepareStmt: PreparedStatement = null
    val inputPath = "/Users/lijiacen/Downloads/test.txt"
    env.readTextFile(inputPath).map(m => {
      var arr = m.split(" ")
      (arr(0), Integer.parseInt(arr(1)))
    }).addSink(new RichSinkFunction[(String, Int)] {
      @throws[Exception]
      override def invoke(value: Tuple2[String, Int], context: SinkFunction.Context): Unit = {
        println(value._1, value._2)
        prepareStmt = connection.prepareStatement("insert into student(name,age) values(?,?)");
        prepareStmt.setString(1, value._1);
        prepareStmt.setInt(2, value._2)
        prepareStmt.execute();
      }

      override def close(): Unit = {
        MysqlUtils.closeConnetion(connection)
      }

      override def open(parameters: Configuration): Unit = {
        connection = MysqlUtils.getConnection()
      }
    })


    env.execute("Sink2Mysql")
  }
}
